💡 導讀：考前衝刺時，你是每次都重讀整本書，還是直接看「錯題本」？

當你準備學測或段考時，遇到不會的題目，你是每一次都花 20 分鐘把整本 500 頁的厚重教科書從第一頁開始重新翻一遍，還是打開你自己整理好的 「精華錯題本」，在 3 秒鐘內直接找到答案？

答案顯然是後者！

在真實醫院的臨床基因診斷中也是完全一樣的道理。當我們每天讀取不同病患的基因檔案時，很多常見或歷史查詢過的變異（SNV）其實會重複出現。如果每一次遇到相同的變異，都傻傻地重新向 Google 1 PB 資料庫發送網路請求，不僅會浪費 API 額度與時間，系統也會變得很慢。

今天，我們就要進入模組四的最終壓軸（Day 20）！我們要教大家如何將前面幾天寫好的 Python 模組打包，實作一套完整的 「VCF 檔自動化過濾與快取管線 (VCF Filter & Cache Pipeline)」！

這套管線能自動解析病患的 VCF 基因檔案，並建立像「錯題本」一樣的快取資料庫（Cache）——查過的變異 0.001 秒直接回傳，沒查過的才向雲端查詢並自動存入，打造真正 Production（正式營運）級的生醫數據系統！

一、 什麼是 VCF 檔？病患基因差異的「身分證總清冊」

在動手寫程式之前，我們必須先認識生醫資訊界最核心的檔案格式——VCF (Variant Call Format)。

當醫院對病患進行全基因組測序（WGS）後，檢測機器不會把 30 億個字母全部塞給你，而是只會把病患與標準人類基因組（hg38）**「不一樣的地方」**印成一張清單，這張清單就是 VCF 檔案。

一份標準的 VCF 檔案裡面包含數百萬列數據，格式長這樣：

#CHROM   POS         ID   REF  ALT  QUAL  FILTER  INFO
chr9     128225994   .    G    A    99    PASS    DP=45;AF=0.5

• CHROM (染色體)：chr9（第 9 號染色體）
• POS (座標位置)：128225994（門牌號碼）
• REF / ALT (野生型/突變型)：原本是 G，病患突變成 A（這正是我們前幾天拆解的 DNM1 致命變異！）
• INFO (附加資訊)：包含測序深度（DP）與突變頻率（AF）等品質指標。

我們的第一步，就是用 Python 自動讀取這個 VCF 身分證總清冊，並過濾掉品質不佳的雜訊！

二、 為什麼需要快取（Cache）？快取 Hit vs. Cache Miss

什麼是快取（Cache）？簡單來說，快取就是一個「記憶快貼簿」或「本地小資料庫」（例如 SQLite 或雲端 Firebase）。

當我們的系統接到一個變異查詢請求時，會執行以下兩步判斷：

1. 快取命中 (Cache HIT ⚡)：
   系統先去本地快取資料庫查：「這個變異以前有人查過嗎？」如果查到了，直接從本地拿出結果，耗時不到 0.001 秒，完全不消耗 API 額度！
2. 快取未命中 (Cache MISS ☁️)：
   如果本地資料庫沒看過這個變異，系統才會透過 asyncio 發送請求給 AlphaGenome API，撈回 AVI 分數與多模態軌跡後，立刻將結果「寫入快取」，讓下一次查詢變得超快！

💻 三、 實作藍圖：自動化 VCF 過濾與快取管線 (vcf_cache_pipeline.py)

打開程式碼編輯器，把前幾天的「非同步批次查詢」與「快取機制」結合，寫出這套 Production 級的數據管線：

import os
import json
import asyncio
from dotenv import load_dotenv
import alphagenome as ag

# 載入隱藏設定檔中的憑證
load_dotenv()

# 本地模擬快取檔案 (JSON 快取資料庫)
CACHE_FILE = "gene_cache.json"

def load_cache():
    """載入本地快取資料庫"""
    if os.path.exists(CACHE_FILE):
        with open(CACHE_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

def save_cache(cache_data):
    """更新並寫入本地快取資料庫"""
    with open(CACHE_FILE, "w", encoding="utf-8") as f:
        json.dump(cache_data, f, ensure_ascii=False, indent=2)

async def process_vcf_entry(client, variant_str, cache):
    """處理單一 VCF 變異：先查快取，若無才呼叫 API"""
    # 1. 檢查快取命中 (Cache HIT)
    if variant_str in cache:
        print(f"⚡ [Cache HIT] 變異 {variant_str} 已存在快取中！秒速回傳結果。")
        return cache[variant_str]

    # 2. 快取未命中 (Cache MISS)：發送 API 查詢
    print(f"☁️ [Cache MISS] 變異 {variant_str} 為全新變異，正在向 AlphaGenome API 查詢...")
    try:
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(None, client.get_variant_annotation, variant_str)

        avi_score = response.get('avi_score', {}).get('percentile', 0)
        top_drivers = response.get('feature_attribution', {}).get('top_drivers', [])

        # 打包要快取的結構化結果
        result_data = {
            "variant_id": variant_str,
            "avi_score": avi_score,
            "top_driver": top_drivers['feature_name'] if top_drivers else "Unknown",
            "status": "SUCCESS"
        }

        # 寫入快取記憶體
        cache[variant_str] = result_data
        return result_data

    except Exception as e:
        print(f"❌ 變異 {variant_str} 查詢失敗: {e}")
        return {"variant_id": variant_str, "status": "FAILED", "avi_score": 0}

async def run_vcf_pipeline(vcf_variants_list):
    """主管線：讀取 VCF 變異清單，批次處理並自動更新快取"""
    client = ag.Client()
    cache = load_cache()

    print(f"🚀 啟動 VCF 自動化過濾與快取管線，準備處理 {len(vcf_variants_list)} 個變異...\n")

    tasks = [process_vcf_entry(client, var, cache) for var in vcf_variants_list]
    results = await asyncio.gather(*tasks)

    # 將最新查詢到的新變異資料存回硬碟 JSON 檔
    save_cache(cache)
    print("\n💾 快取資料庫已同步更新完畢！")

    # 按 AVI 分數進行高到低排序 (Top Prioritization)
    sorted_results = sorted(results, key=lambda x: x.get('avi_score', 0), reverse=True)
    return sorted_results

if __name__ == "__main__":
    # 模擬從病患 VCF 檔案讀取出的 5 個候選變異
    # （包含前幾天我們熟知的 DNM1 變異與其它模擬變異）
    vcf_input = [
        "chr9:128225994:G:A",  # DNM1 致命變異
        "chr9:128225900:A:G",
        "chr9:128225910:C:T",
        "chr9:128225994:G:A",  # 重複出現的變異 (測試 Cache HIT)
        "chr1:45678900:T:C"
    ]

    # 第一次執行：會進行 API 查詢並建立快取
    print("--- 第一次執行管線 (建立快取) ---")
    first_run = asyncio.run(run_vcf_pipeline(vcf_input))

    print("\n" + "="*50)
    print("🏆 病患 VCF 最終排序 Top 高風險變異報告：")
    for rank, res in enumerate(first_run[:3], 1):
        print(f"  #{rank} {res['variant_id']} ➔ AVI 分數: {res['avi_score']} | 主要機制: {res['top_driver']}")
    print("="*50 + "\n")

    # 第二次執行：測試快取秒速回傳 (全數 Cache HIT)
    print("--- 第二次執行管線 (測試快取命中) ---")
    second_run = asyncio.run(run_vcf_pipeline(vcf_input))

執行這段腳本，你會發現第二次執行時，系統連一次網路 API都沒呼叫，就在 0.001 秒內瞬間印出排序結果！這就是 Production 級資料管線的威力！